package com.rabbitmq.rabbitmq.receiver;

import com.caicai.domain.User;
import com.rabbitmq.client.Channel;
import com.rabbitmq.rabbitmq.config.RpcMessageConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

@Component
public class ReceiverListener {
	ExecutorService executor = Executors.newFixedThreadPool(5);

	@RabbitListener(queues = {"${mq.quenue.string}"})
	public  void stringMessage(String str, Channel channel, Message message) throws IOException {

		try {

			//实验结果：监听队列为同步，如测试1 ，需要等待10秒才会进来一条消息，basicQos 设置无效，采用异步方式，每隔10秒获取两个消息，basicQos生效，
			// 注意RabbitListenerContainerFactory的配置，在配置文件中不生效，原因不明，需要写bean的配置

			/*测试1
			channel.basicQos(0,2,false);
			System.out.println("接收到消息"+message);
			channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
			Thread.sleep(10000);*/
			Future<String> f = executor.submit(new Callable<String>() {

				@Override
				public String call() throws Exception {
					channel.basicQos(0,2,false);
					System.out.println("接收到消息"+message);
					Thread.sleep(10000);
					channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
					return "hello";
				}
			});
			//消息消费成功手动确认（防止消息丢失）
			//deliveryTag:该消息的index
			//multiple：是否批量. true：将一次性ack所有小于deliveryTag的消息
			//System.out.println("1111");
		} catch (Exception e) {
			//消费消息失败处理
			//deliveryTag:该消息的index。
			//multiple：是否批量. true：将一次性拒绝所有小于deliveryTag的消息。
			//requeue：被拒绝的是否重新入队列。（若是为true如果消费者在处理过程中一直抛出异常，会导致入队-》拒绝-》入队的循环）
			//思考：可根据异常类型决定是否重新发送，也可重新在此发送消息进入消息队列
			channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);

		}

	}

	@RabbitListener(queues = {"${mq.quenue.user}"})
	public  void userMessage(Message user){

		System.out.println("接收到消息"+user);
	}

	@RabbitListener(queues = RpcMessageConfig.rpcQueue)
	public String rpcmMessage(String  message){
		System.out.println("接收到消息------"+message);
		return "回调rpc消息";
	}

	@RabbitListener(queues =RpcMessageConfig.replyQueue)
	public void rpcResponseMessage(String  message){
		System.out.println("接收到消息1111"+message);

	}

}
